package com.roncoo.eshop.datasync.rabbitmq;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.locks.ReentrantLock;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.roncoo.eshop.datasync.service.ProductService;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

/**
 * 接收源数据通过rabbitmq发送的变更消息 修改redis缓存 发送下一个队列，提供给数据聚合服务
 * 
 * @author lileping
 */
@Component
@RabbitListener(queues = "data-change-queue")
public class DataChangeQueueReceiver {
	
	// 接收消息放入该set进行限时去重
	private Set<String> dimDataChangeMessageSet = new HashSet<String>();
	// 待发送的消息集合（将要发送的数据从dim set中交换到该set中，线程发送；原dim set 继续接收消息队列消息）
	private Set<String> toSendMessageSet = new HashSet<String>();
	
	// 品牌批次发送消息
	private List<JSONObject> brandBatchMessageList = new ArrayList<JSONObject>();
	
	@Autowired
	private ProductService productService;
	
	@Autowired
	private JedisPool jedisPool;

	@Autowired
	private RabbitmqSender rabbitmqSender;
	
	public DataChangeQueueReceiver() {
		// 启动发送消息的线程
		new SendThread().start();
	}
	
	/**
	 * 获取锁，操作去重消息队列
	 */
	public ReentrantLock lock = new ReentrantLock();

	@RabbitHandler
	public void process(String message) {
		try {
			JSONObject messageObject = JSONObject.parseObject(message);
			String dataType = messageObject.getString("data_type");
			if ("brand".equals(dataType)) {
				processBrandDataChangeMessage(messageObject);
			} else if ("category".equals(dataType)) {
				processCategoryDataChangeMessage(messageObject);
			} else if ("product_intro".equals(dataType)) {
				processProductIntroDataChangeMessage(messageObject);
			} else if ("product".equals(dataType)) {
				processProductDataChangeMessage(messageObject);
			} else if ("product_property".equals(dataType)) {
				processProductPropertyDataChangeMessage(messageObject);
			} else if ("product_specification".equals(dataType)) {
				processProductSpecificationDataChangeMessage(messageObject);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 添加消息到去重队列中
	 * @param message
	 */
	public void addMessageToSet(String message) {
		lock.lock();
		try {
			// 将消息发送到聚合服务
			dimDataChangeMessageSet.add(message);
		} finally{
			lock.unlock();
		}
	}

	/**
	 * 处理品牌数据变更 
	 * @param messageJson
	 */
	public void processBrandDataChangeMessage(JSONObject messageJson) {
		Long id = messageJson.getLong("id");
		String eventType = messageJson.getString("event_type");
		Jedis jedis = jedisPool.getResource();
		if ("add".equals(eventType) || "update".equals(eventType)) {

			brandBatchMessageList.add(messageJson);
			// 大于3条批量处理 (但是如果有数据不是很频繁，只有两条或1条数据，可能会导致没有处理到)
			if(brandBatchMessageList.size() > 3) {
				String ids = "";
				for (int i = 0 ; i < brandBatchMessageList.size(); i++) {
					JSONObject message = brandBatchMessageList.get(i);
					ids += message.getLong("id");
					if(i < brandBatchMessageList.size() - 1) {
						ids += ",";
					}
				}
				
				JSONArray  messageJsonArr = JSONArray.parseArray(productService.findByIds(ids));
				for (int i = 0 ; i < messageJsonArr.size(); i++) {
					// 查询源服务数据
					JSONObject dataJson = messageJsonArr.getJSONObject(i);
					Long brandId = dataJson.getLong("id"); 
					// 更新redis缓存
					jedis.set("brand_" + brandId, JSONObject.toJSONString(dataJson));
					
					// 将消息发送到聚合服务
					addMessageToSet("{\"dim_type\" : \"brand\", \"id\" : " + brandId + "}");
				}
				brandBatchMessageList.clear();
			}
		} else {
			// 删除缓存
			jedis.del("brand_" + id);
		}
		
	}
	
	/**
	 * 处理商品类型数据变更消息
	 * @param messageJSONObject
	 */
	private void processCategoryDataChangeMessage(JSONObject messageJSONObject) {
    	Long id = messageJSONObject.getLong("id"); 
    	String eventType = messageJSONObject.getString("event_type"); 
    	
    	if("add".equals(eventType) || "update".equals(eventType)) { 
    		JSONObject dataJSONObject = JSONObject.parseObject(productService.findCategoryById(id));  
    		Jedis jedis = jedisPool.getResource();
    		jedis.set("category_" + dataJSONObject.getLong("id"), dataJSONObject.toJSONString());
    	} else if ("delete".equals(eventType)) {
    		Jedis jedis = jedisPool.getResource();
    		jedis.del("category_" + id);
    	}
    	  
    	addMessageToSet("{\"dim_type\": \"category\", \"id\": " + id + "}");
    }
    
	/**
	 * 处理商品详细信息变更消息
	 * @param messageJSONObject
	 */
    private void processProductIntroDataChangeMessage(JSONObject messageJSONObject) {
    	Long id = messageJSONObject.getLong("id"); 
    	Long productId = messageJSONObject.getLong("product_id");
    	String eventType = messageJSONObject.getString("event_type"); 
    	
    	if("add".equals(eventType) || "update".equals(eventType)) { 
    		JSONObject dataJSONObject = JSONObject.parseObject(productService.findProductIntroById(id));  
    		Jedis jedis = jedisPool.getResource();
    		jedis.set("product_intro_" + productId, dataJSONObject.toJSONString());
    	} else if ("delete".equals(eventType)) {
    		Jedis jedis = jedisPool.getResource();
    		jedis.del("product_intro_" + productId);
    	}
    	  
    	addMessageToSet("{\"dim_type\": \"product_intro\", \"id\": " + productId + "}");
    }
    
    /**
     * 处理商品基本信息变更
     * @param messageJSONObject
     */
    private void processProductDataChangeMessage(JSONObject messageJSONObject) {
    	Long id = messageJSONObject.getLong("id"); 
    	String eventType = messageJSONObject.getString("event_type"); 
    	
    	if("add".equals(eventType) || "update".equals(eventType)) { 
    		JSONObject dataJSONObject = JSONObject.parseObject(productService.findProductById(id));  
    		Jedis jedis = jedisPool.getResource();
    		jedis.set("product_" + id, dataJSONObject.toJSONString());
    	} else if ("delete".equals(eventType)) {
    		Jedis jedis = jedisPool.getResource();
    		jedis.del("product_" + id);
    	}
    	  
    	addMessageToSet("{\"dim_type\": \"product\", \"id\": " + id + "}");
    }
    
    private void processProductPropertyDataChangeMessage(JSONObject messageJSONObject) {
    	Long id = messageJSONObject.getLong("id"); 
    	Long productId = messageJSONObject.getLong("product_id");
    	String eventType = messageJSONObject.getString("event_type"); 
    	
    	if("add".equals(eventType) || "update".equals(eventType)) { 
    		JSONObject dataJSONObject = JSONObject.parseObject(productService.findProductPropertyById(id));  
    		Jedis jedis = jedisPool.getResource();
    		jedis.set("product_property_" + productId, dataJSONObject.toJSONString());
    	} else if ("delete".equals(eventType)) {
    		Jedis jedis = jedisPool.getResource();
    		jedis.del("product_property_" + productId);
    	}
    	  
    	addMessageToSet("{\"dim_type\": \"product\", \"id\": " + productId + "}");
    }
    
    private void processProductSpecificationDataChangeMessage(JSONObject messageJSONObject) {
    	Long id = messageJSONObject.getLong("id"); 
    	Long productId = messageJSONObject.getLong("product_id");
    	String eventType = messageJSONObject.getString("event_type"); 
    	
    	if("add".equals(eventType) || "update".equals(eventType)) { 
    		JSONObject dataJSONObject = JSONObject.parseObject(productService.findProductSpecificationById(id));  
    		Jedis jedis = jedisPool.getResource();
    		jedis.set("product_specification_" + productId, dataJSONObject.toJSONString());
    	} else if ("delete".equals(eventType)) {
    		Jedis jedis = jedisPool.getResource();
    		jedis.del("product_specification_" + productId);
    	}
    	  
    	addMessageToSet("{\"dim_type\": \"product\", \"id\": " + productId + "}");
    }
    
    /**
     * 把待发送的消息，从队列中交换出来
     */
    public void swapToSendSet() {
    	try {
    		lock.lock();
    		toSendMessageSet.addAll(dimDataChangeMessageSet);
        	dimDataChangeMessageSet.clear();
    	} finally {
    		lock.unlock();
    	}
    }
    
    /**
     * 发送消息
     * @author lileping
     */
    class SendThread extends Thread{
    	@Override
    	public void run() {
    		while(true) {
    			try {
    				// 把去重队列的数据，交换出来，进行发送
    				swapToSendSet();
    				if(toSendMessageSet.size() > 0) {
    	    			for (String message : toSendMessageSet) {
    	    				try {
    	    					rabbitmqSender.send("data-aggr-change-queue", message);
    	    				} catch (Exception e) {
    	    					e.printStackTrace();
    	    					// 异常消息，可计数进入异常队列; 异常消息需要消息确认; 同时要考虑缓存这种场景，有定时过期策略，是否需要有必要使用这种ack机制
    	    				}
    					}
    	    		}
    				toSendMessageSet.clear();
    	    		// 5秒钟发送一次
    	    		Thread.sleep(5000);
    			} catch(Exception e) {
    				e.printStackTrace();
    			}
    		}
    		
    	}
    }



}
